package sc.rawdata;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.io.Text;
import org.hibernate.Query;
import org.hibernate.Session;

import sc.hibernate.config.HibernateSession;
import sc.hibernate.models.Quote;
import sc.writable.PriceData;
import sc.writable.WritableArrayListFloat;
import sc.writable.WritableArrayListInteger;

/**
 * Methods to get and update the informations from EOD Data
 * 
 * @author Quentin Rossé
 *
 */
public class UpdateEodData {

	/**
	 * price data loaded from the file
	 */
	private static TreeMap<String, PriceData> priceData;
	
	/**
	 * Number max of price data in the sequence file
	 */
	private static final int MAX_DATA = 250;
	
	
	/**
	 * Reads a file with eoddata format and returns the informations
	 * 
	 * @param file the eoddata file
	 * @return informations about the stocks
	 * @throws IOException
	 */
	public static ArrayList<String[]> readEodFile(File file) throws IOException{
		FileReader fr = new FileReader(file);
		BufferedReader br = new BufferedReader(fr);
		String[] stockData;
		ArrayList<String[]> data = new ArrayList<String[]>();
		       
		while(br.ready()){
			stockData = br.readLine().split(",");
			
			data.add(stockData);
		}
		
		return data;
	}
	
	
	
	/**
	 * Read the price file and return the price data
	 * 
	 * @param path path to the price file
	 * @param conf defines hdfs properties (new Configuration() for local fs)
	 * 
	 * @return price data : key = stockExchange:stock
	 *                      value = WritableArrayListFloat of the prices
	 * 
	 * @throws IOException
	 */
	public static TreeMap<String, PriceData> readPriceFile(String path,
            Configuration conf) 
		throws IOException{
		
		FileSystem fs = FileSystem.get(conf);
		
		FileStatus[] files = fs.listStatus(new Path(path));
		
		TreeMap<String, PriceData> priceData = 
			new TreeMap<String, PriceData>(); 
		
		Text key = new Text();
		PriceData value = new PriceData();
		
		if(files != null)
			for(FileStatus file : files){
				if(!file.isDir()){
					SequenceFile.Reader reader = new SequenceFile.Reader(fs, file.getPath(), conf);
				    
				    while(reader.next(key, value)){
				    	priceData.put(key.toString(), value);
				    	value = new PriceData();
				    }
				    
				    reader.close();
				}
			}

		return priceData;
	}
	
	
	
	private static void updatePriceData(String stock, float newClose, float newLow, float newHigh, int newVolume){
		if(priceData.containsKey(stock)){
			priceData.get(stock).getClosePrices().add(newClose, MAX_DATA);
			priceData.get(stock).getDayLowPrices().add(newLow, MAX_DATA);
			priceData.get(stock).getDayHighPrices().add(newHigh, MAX_DATA);
			priceData.get(stock).getVolume().add(newVolume, MAX_DATA);
		}
		else{
			WritableArrayListFloat walfClose = new WritableArrayListFloat();
			WritableArrayListFloat walfLow = new WritableArrayListFloat();
			WritableArrayListFloat walfHigh = new WritableArrayListFloat();
			WritableArrayListInteger walVolume = new WritableArrayListInteger();
			walfClose.add(newClose);
			walfLow.add(newLow);
			walfHigh.add(newHigh);
			walVolume.add(newVolume);
			priceData.put(stock, new PriceData(walfClose, walfLow, walfHigh, walVolume));
		}
		//System.out.println(stock + " : " + priceData.get(stock).getClosePrices());		
	}
	
	/**
	 * 
	 * @param args  [0] : HDFS config file
	 * 				[1] : sequence datafile
	 * @throws IOException
	 */
	public static void main(String[] args) throws IOException{
		System.out.println("HDFS config file : " + args[0]);
		System.out.println("dataFile : " + args[1]);
		
		Configuration hdfsConf = new Configuration();
		hdfsConf.addResource(new Path(args[0]));
		
		//Configuration localConf = new Configuration();
		
		priceData = readPriceFile(args[1], hdfsConf);
		
		
		//Get the data from DB and update the price data
		Session session = HibernateSession.getSessionFactory().openSession();
		
		Date date = new Date();
		Query hql = session.createQuery("SELECT SYMBOL,PRICE,DAILY_LOW,DAILY_HIGH,VOLUME FROM Quote WHERE date = '" + date+"'");
		List<?> quotes = hql.list();
		session.getTransaction().commit();
		
		for(int i=0; i < quotes.size(); i++){
			Quote quote = (Quote) quotes.get(i);
			updatePriceData(quote.getCompany().getSymbol(), 
			        quote.getDaily_low(), 
			        quote.getDaily_high(),
			        quote.getPrice(),
			        quote.getVolume());
		}

		//write the new data to the file
		Writer seqWriter =
			SequenceFile.createWriter(FileSystem.get(hdfsConf), 
					                  hdfsConf, 
					                  new Path(args[1]), 
					                  Text.class, 
					                  PriceData.class);
		Text key = new Text();
		Iterator<String> keys = priceData.keySet().iterator();
		while(keys.hasNext()){
			key.set(keys.next());
			seqWriter.append(key, priceData.get(key.toString()));
		}
		
		seqWriter.close();
	}
}
